package org.example;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class PaimonDDL_ColumnAlter {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT,"8081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(conf);
        env.setParallelism(1);
        TableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);

        tableEnv.executeSql("CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");\n");

//        tableEnv.executeSql("CREATE TABLE IF NOT EXISTS `paimon`.`default`.`test_alter_column` (\n" +
//                "                user_id BIGINT NOT NULL,\n" +
//                "                item_id BIGINT,\n" +
//                "                behavior STRING,\n" +
//                "                dt STRING NOT NULL,\n" +
//                "                hh STRING NOT NULL,\n" +
//                "                proc_time AS PROCTIME(),\n" +
//                "                PRIMARY KEY (dt, hh, user_id) NOT ENFORCED\n" +
//                "                );");
//

//        tableEnv.executeSql("alter table `paimon`.`default`.`test_alter_column` drop (id);").print();
//        tableEnv.executeSql("desc `paimon`.`default`.`test_alter_column`;").print();

//        tableEnv.executeSql("ALTER TABLE `paimon`.`default`.`test_alter_column` MODIFY description string FIRST;");

//        tableEnv.executeSql("DROP TABLE IF EXISTS `paimon`.`default`.`test_alter_watermark_column`");
//
//        tableEnv.executeSql("CREATE TABLE IF NOT EXISTS `paimon`.`default`.`test_alter_watermark_column` (\n" +
//                "                user_id BIGINT NOT NULL,\n" +
//                "                item_id BIGINT,\n" +
//                "                behavior STRING,\n" +
//                "                dt STRING NOT NULL,\n" +
//                "                hh STRING NOT NULL,\n" +
//                "                create_time STRING NOT NULL,\n" +
//                "                PRIMARY KEY (dt, hh, user_id) NOT ENFORCED\n" +
//                "                );");
//
//        tableEnv.executeSql("desc `paimon`.`default`.`test_alter_watermark_column`;").print();
//
//        tableEnv.executeSql("ALTER TABLE `paimon`.`default`.`test_alter_watermark_column` ADD(\n" +
//                "    proc_time AS PROCTIME() \n" +
//                ");");

        tableEnv.executeSql("ALTER TABLE `paimon`.`default`.`test_alter_watermark_column` DROP WATERMARK;");

        tableEnv.executeSql("desc `paimon`.`default`.`test_alter_watermark_column`;").print();
    }
}
